use super::{handler::Consumer, Connection, Handler};
use crate::{
    cache::{Metadata, Topic},
    Argument, Error, Token,
};
use log::{error, info};
use std::fmt::Write;
use std::sync::Arc;
use tokio::sync::Mutex;

pub async fn produce_command(
    token: Token,
    mut conn: Connection,
    metadata: Arc<Mutex<Metadata>>,
    handler: &Handler,
) -> Result<(), Error> {
    let mut produce_topic = None;
    let mut partition = None;
    for arg in token.args {
        match arg {
            Argument::Topic { topic } => {
                produce_topic = Some(topic);
            }
            Argument::Partitions { partitions } => {
                partition = Some(partitions);
            }
            _ => {}
        }
    }

    if let Some(topic) = produce_topic {
        handler.produce_message(&topic, partition, conn).await?;
    }

    Ok(())
}

pub async fn delete_topic_command(
    token: Token,
    mut conn: Connection,
    metadata: Arc<Mutex<Metadata>>,
) -> Result<(), Error> {
    let mut delete_topic = None;
    for arg in token.args {
        match arg {
            Argument::Topic { topic } => {
                delete_topic = Some(topic);
            }
            _ => {}
        }
    }

    if let Some(topic) = delete_topic {
        let mut metadata_lock = metadata.lock().await;
        metadata_lock.topic_map.remove(&topic);
    }
    Ok(())
}

pub async fn list_command(
    mut conn: Connection,
    metadata: Arc<Mutex<Metadata>>,
) -> Result<(), Error> {
    conn.buf.clear();
    if let Err(e) = conn
        .buf
        .write_str("topic_name\tpartition_nun\treplication_factor\n")
    {
        info!("command list error, {}", e);
    }
    conn.write_message().await?;

    let metadata_lock = metadata.lock().await;

    for (_k, v) in metadata_lock.topic_map.iter() {
        if let Err(e) = conn.buf.write_str(&format!(
            "{}\t{}\t{}\n",
            v.topic_name, v.partition_num, v.replication_factor
        )) {
            info!("command list error, {}", e);
        }
        conn.write_message().await?;
    }
    Ok(())
}

pub async fn create_topic_command(
    token: Token,
    mut conn: Connection,
    metadata: Arc<Mutex<Metadata>>,
) -> Result<(), Error> {
    let mut topic_cache: Topic = Topic::default();
    for arg in token.args {
        match arg {
            Argument::Topic { topic } => topic_cache.topic_name = topic,
            Argument::Partitions { partitions } => topic_cache.partition_num = partitions,
            Argument::ReplicationFactor { replication_factor } => {
                topic_cache.replication_factor = replication_factor
            }
            _ => {}
        }
    }
    {
        let mut metadata_lock = metadata.lock().await;
        metadata_lock
            .topic_map
            .insert(topic_cache.topic_name.clone(), topic_cache);
    }
    Ok(())
}

pub async fn consume_command(
    token: Token,
    mut conn: Connection,
    metadata: Arc<Mutex<Metadata>>,
) -> Result<(), Error> {
    let mut consume_topic = None;
    let mut partition = None;
    for arg in token.args {
        match arg {
            Argument::Topic { topic } => {
                consume_topic = Some(topic);
            }
            Argument::Partitions { partitions } => {
                partition = Some(partitions);
            }
            _ => {}
        }
    }

    if let None = consume_topic {
        conn.buf.clear();
        if let Err(_) = conn.buf.write_str("Please configure the consume topic\n") {
            error!("Please configure the consume topic");
        }
        conn.write_message().await?;
        return Ok(());
    }

    if let Some(topic) = consume_topic {
        {
            let metadata_lock = metadata.lock().await;
            let topic_map = metadata_lock.topic_map.get(&topic);
            if let None = topic_map {
                conn.buf.clear();
                if let Err(_) = conn.buf.write_str("Please configure the create topic\n") {
                    error!("Please configure the create topic");
                }
                conn.write_message().await?;
                return Ok(());
            }
        }
        let mut partitions = Vec::new();
        match partition {
            Some(partition) => {
                partitions.push(partition);
            }
            None => {
                let metadata_lock = metadata.lock().await;
                let topic_map = metadata_lock.topic_map.get(&topic);
                if let Some(topic) = topic_map {
                    for partition in 0..topic.partition_num {
                        partitions.push(partition);
                    }
                }
            }
        };
        let mut consumer = Consumer::new(topic, partitions, conn).await;
        consumer.send_message().await?;
    }
    Ok(())
}
